[AWS IoT] 既存の証明書だけでMQTT以外の各種AWSリソ−スにアクセスする (Authorizing Direct Calls)

AWS IoT の Authorizing Direct Callsを使用すると、既存の証明書だけで、AWSリソースにアクセスする事が可能です。

#AWS IoT Core

#AWS

SIN

2020.08.29

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

1 はじめに

CX事業本部の平内（SIN）です。

AWS IoT のメッセージコアへのMQTTアクセスでは、通常、X.509 証明書が使用される場面が多いと思います。

しかし、例えば、「デバイスからMQTT以外に、S3への画像送信が必要」となった場合、ちょっと思いつくのは、以下のような感じでしょうか・・・

MQTTのペイロードに画像を押し込んで、ルールで処理する（MQTTのサイズ制限がちょっと不安）
ユーザIDを発行してアクセスキーをデバイスに配置（ちょっと、美しくない？）
CognitoのIdentity IDで、RoleからMQTTとS3の権限付与する（せっかく証明書があるのに・・・）

色々、方法はあると思うのですが、今回は、Authorizing Direct Callsにより、既存の証明書だけで、AWSリソースにアクセスする方法を試してみました。

AWS IoTには、X.509証明書を一意のデバイスIDとして利用でできる認証情報プロバイダーがあり、ここから、一時的なアクセスキー、シークレットキー及び、トークンを取得して、AWS Signature Version 4でAWSの各種リソースにアクセスすることができます。

AWS のサービスの直接呼び出しの認証より

2 手順

既にモノにアタッチされた証明書がレジストされており、証明書、秘密鍵及び、ルートCAが、デバイス側に配置されている状態だとすると、設定が必要なのは、以下の３つです。（かっこ内の名前は、今回、作業に使用した名前です）

最終的に付与したいRoleの作成（ADCTestRole）
ロールエイリアスの作成（ADCTestRoleAlias）
証明書に紐付くポリシーへ権限追加（ADCTestPolicy）

また、証明書を利用してAWSリソースにアクセスする手順は、以下のようになります。

一時クレデンシャルのエンドポイントを取得
エンドポイントから一時クレデンシャルを取得
取得した一時クレデンシャルを使用してAWSリソースにアクセス

以下、今回作業したリソースの名前です。

モノ: ADCTestThing
ロール: ADCTestRole
ロールエイリアス: ADCTestRoleAlias
ポリシー:ADCTestPolicy

3 設定

(1) 最終的に付与したいRoleの作成

最初に、一時クレデンシャルに付与するロール（ここでは、ADCTestRoleとしました）を作成します。

下記は、S3のバケット(adc-test-2020-08-26)にアクセスするためのポリシーの例です。

Permissions policies

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::adc-test-2020-08-26/*",
                "arn:aws:s3:::adc-test-2020-08-26"
            ]
        }
    ]
}

また、このロールは、AWS IoTの（資格情報）サービスからAssumeRoleされるので、信頼元をcredentials.iot.amazonaws.comとします。

Trust Relationship

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "credentials.iot.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

(2) ロールエイリアスの作成

上記で作成したロールへのエイリアス作成です。

AWS IoTのコンソールから、安全性 > ロールエイリアス　と辿り、新しくロールエイリアスを作成します。（ここでは、名前をADCTestRoleAliasとしました）

元となるロールは、選択ボタンを押すと、AWS IoTに信頼関係を付与されているロールが一覧されますので、上で作成したADCTestRoleを選択します。

(3)証明書に紐付くポリシーへ権限追加

使用する証明書には、上記で作成したロールエイリアスへの iot:AssumeRoleWithCertificateが必要です。既に、ポリシーにMQTTへのアクセス等の権限が付与されている場合は、iot:AssumeRoleWithCertificateを追加することになります。

ADCTestPolicy

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:AssumeRoleWithCertificate",
      "Resource": "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:rolealias/ADCTestRoleAlias"
    }
  ]
}

4 使用方法

ここまでで設定は完了です。証明書を使用してAWSリソースにアクセスする手順は、下記のとおりです。

一時クレデンシャルのエンドポイントを取得
エンドポイントから一時クレデンシャルを取得
取得した一時クレデンシャルを使用してAWSリソースにアクセス

(1) 一時クレデンシャルのエンドポイントを取得

% aws iot describe-endpoint --endpoint-type iot:CredentialProvider --p profileName
{
    "endpointAddress": "xxxxx.credentials.iot.ap-northeast-1.amazonaws.com"
}

(2) エンドポイントから一時クレデンシャルを取得

% curl --cert ./cert.pem --key ./private.key -H "x-amzn-iot-thingname: ADCTestThing" https://xxxxx.credentials.iot.ap-northeast-1.amazonaws.com/role-aliases/ADCTestRoleAlias/credentials | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  1289  100  1289    0     0   4131      0 --:--:-- --:--:-- --:--:--  4131
{
  "credentials": {
    "accessKeyId": "ASIAWMOBC4JXPPGJBPW5",
    "secretAccessKey": "UEjTxv84aLhSXp1p/Z4jqsO5zXClvNGwBbtAlMA1",
    "sessionToken": "IQoJbxxxxxxxxxxxxxxxxxNm5Yofg==",
    "expiration": "2020-08-29T02:02:12Z"
  }
}

(3) 取得した一時クレデンシャルを使用してAWSリソースにアクセス

% export AWS_ACCESS_KEY_ID=ASIAWMOBC4JXPPGJBPW5
% export AWS_SECRET_ACCESS_KEY=UEjTxv84aLhSXp1p/Z4jqsO5zXClvNGwBbtAlMA1
% export AWS_SESSION_TOKEN=IQoJbxxxxxxxxxxxxxxxxxNm5Yofg==
% aws s3 cp s3://adc-test-2020-08-26/test.txt .
download: s3://adc-test-2020-08-26/test.txt to ./test.txt

上記を、Pythonで書いた例です。

import requests
import json
import boto3
from boto3.session import Session

endpoint = "https://xxxxx.credentials.iot.ap-northeast-1.amazonaws.com"
role_alias = 'ADCTestRoleAlias'

result = requests.get(
    f'{endpoint}/role-aliases/{role_alias}/credentials',                                                                                                                                                        
    cert=(f'./cert.pem', f'./private.key')
)

if(result.status_code != 200):
    exit()

body = json.loads(result.text)
access_key = body["credentials"]["accessKeyId"]
secret_key = body["credentials"]["secretAccessKey"]
token = body["credentials"]["sessionToken"]

session = Session(aws_access_key_id=access_key,
                aws_secret_access_key=secret_key,
                aws_session_token=token)

s3 = session.resource('s3')
bucket_name = 'adc-test-2020-08-26'
file = 'test.txt'
bucket = s3.Bucket(bucket_name)
bucket.download_file(file, file)